package com.zyb.insurance.policy.infrastructure;

import com.alibaba.fastjson.JSONObject;
import com.zyb.insurance.common.HBaseClient;
import com.zyb.insurance.policy.entity.InsurancePolicy;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.springframework.stereotype.Component;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * hbase管理组件
 */
@Component
public class HBaseManager {

    /**
     * 如果hbase保单表不存在的话，则创建该表
     */
    public void createInsurancePolicyTableIfNotExists() {
        try {
            // 每次这个类实例化的时候，也就是系统启动的时候
            // 会去检查一下这个表是否存在，如果不存在，则创建表
            HBaseClient hBaseClient = HBaseClient.getInstance();
            Connection connection = hBaseClient.getConnection();

            Admin admin = connection.getAdmin();
            if(admin.tableExists(TableName.valueOf("insurance_policy"))) {
                System.out.println("hbase保单表已经存在，不需要重复创建......");
                return;
            }

            System.out.println("hbase保单表不存在，即将创建......");
            HTableDescriptor hTableDescriptor = new HTableDescriptor(
                    TableName.valueOf("InsurancePolicy"));
            hTableDescriptor.addFamily(new HColumnDescriptor("typeMode"));
            hTableDescriptor.addFamily(new HColumnDescriptor("insuranceContract"));
            hTableDescriptor.addFamily(new HColumnDescriptor("policyHolder"));
            hTableDescriptor.addFamily(new HColumnDescriptor("insurancePolicyContractStatus"));
            hTableDescriptor.addFamily(new HColumnDescriptor("insurancePlan"));
            hTableDescriptor.addFamily(new HColumnDescriptor("videoMaterials"));
            hTableDescriptor.addFamily(new HColumnDescriptor("changeHistories"));

            admin.createTable(hTableDescriptor);

            System.out.println("hbase保单表已经成功创建......");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 存储保单数据到hbase里去
     */
    public void saveInsurancePolicyIntoHbase(InsurancePolicy insurancePolicy) throws Exception {
        SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd");

        // 获取hbase连接
        HBaseClient hBaseClient = HBaseClient.getInstance();
        Connection connection = hBaseClient.getConnection();

        // 获取hbase里的保单表
        Table table = connection.getTable(TableName.valueOf("InsurancePolicy"));

        // 构建保单put请求
        Put put = new Put(insurancePolicy.getInsurancePolicyId().getBytes());

        put.addColumn("typeMode".getBytes(), "typeMode".getBytes(),
                String.valueOf(insurancePolicy.getTypeMode()).getBytes());

        put.addColumn("insuranceContract".getBytes(), "issueInsurancePolicyOrganization".getBytes(),
                JSONObject.toJSONString(insurancePolicy.getInsuranceContract().getIssueInsurancePolicyOrganization()).getBytes());
        put.addColumn("insuranceContract".getBytes(), "issueInsurancePolicyStaff".getBytes(),
                JSONObject.toJSONString(insurancePolicy.getInsuranceContract().getIssueInsurancePolicyStaff()).getBytes());
        put.addColumn("insuranceContract".getBytes(), "operator".getBytes(),
                insurancePolicy.getInsuranceContract().getOperator().getBytes());
        put.addColumn("insuranceContract".getBytes(), "insuranceCompany".getBytes(),
                JSONObject.toJSONString(insurancePolicy.getInsuranceContract().getInsuranceCompany()).getBytes());
        put.addColumn("insuranceContract".getBytes(), "insuranceProduct".getBytes(),
                JSONObject.toJSONString(insurancePolicy.getInsuranceContract().getInsuranceProduct()).getBytes());
        put.addColumn("insuranceContract".getBytes(), "contractCode".getBytes(),
                insurancePolicy.getInsuranceContract().getContractCode().getBytes());
        put.addColumn("insuranceContract".getBytes(), "relatedContractCode".getBytes(),
                insurancePolicy.getInsuranceContract().getRelatedContractCode().getBytes());
        put.addColumn("insuranceContract".getBytes(), "proposalFormCode".getBytes(),
                insurancePolicy.getInsuranceContract().getProposalFormCode().getBytes());
        put.addColumn("insuranceContract".getBytes(), "payType".getBytes(),
                String.valueOf(insurancePolicy.getInsuranceContract().getPayType()).getBytes());
        put.addColumn("insuranceContract".getBytes(), "saleType".getBytes(),
                String.valueOf(insurancePolicy.getInsuranceContract().getSaleType()).getBytes());
        put.addColumn("insuranceContract".getBytes(), "payPeriod".getBytes(),
                String.valueOf(insurancePolicy.getInsuranceContract().getPayPeriod()).getBytes());
        put.addColumn("insuranceContract".getBytes(), "insurancePolicyYearNumber".getBytes(),
                String.valueOf(insurancePolicy.getInsuranceContract().getInsurancePolicyYearNumber()).getBytes());
        put.addColumn("insuranceContract".getBytes(), "remarkContent".getBytes(),
                insurancePolicy.getInsuranceContract().getRemarkContent().getBytes());
        put.addColumn("insuranceContract".getBytes(), "tags".getBytes(),
                JSONObject.toJSONString(insurancePolicy.getInsuranceContract().getTags()).getBytes());
        put.addColumn("insuranceContract".getBytes(), "insurancePolicyAttendants".getBytes(),
                JSONObject.toJSONString(insurancePolicy.getInsuranceContract().getInsurancePolicyAttendants()).getBytes());

        put.addColumn("policyHolder".getBytes(), "mainBodyType".getBytes(),
                String.valueOf(insurancePolicy.getPolicyHolder().getMainBodyType()).getBytes());
        put.addColumn("policyHolder".getBytes(), "name".getBytes(),
                insurancePolicy.getPolicyHolder().getName().getBytes());
        put.addColumn("policyHolder".getBytes(), "credentialsType".getBytes(),
                String.valueOf(insurancePolicy.getPolicyHolder().getCredentialsType()).getBytes());
        put.addColumn("policyHolder".getBytes(), "credentialsNumber".getBytes(),
                insurancePolicy.getPolicyHolder().getCredentialsNumber().getBytes());
        put.addColumn("policyHolder".getBytes(), "credentialsExpiryDate".getBytes(),
                dateFormat.format(insurancePolicy.getPolicyHolder().getCredentialsExpiryDate()).getBytes());
        put.addColumn("policyHolder".getBytes(), "gender".getBytes(),
                String.valueOf(insurancePolicy.getPolicyHolder().getGender()).getBytes());
        put.addColumn("policyHolder".getBytes(), "birthday".getBytes(),
                dateFormat.format(insurancePolicy.getPolicyHolder().getBirthday()).getBytes());
        put.addColumn("policyHolder".getBytes(), "age".getBytes(),
                String.valueOf(insurancePolicy.getPolicyHolder().getAge()).getBytes());
        put.addColumn("policyHolder".getBytes(), "nationality".getBytes(),
                insurancePolicy.getPolicyHolder().getNationality().getBytes());
        put.addColumn("policyHolder".getBytes(), "nation".getBytes(),
                insurancePolicy.getPolicyHolder().getNation().getBytes());
        put.addColumn("policyHolder".getBytes(), "marriageStatus".getBytes(),
                insurancePolicy.getPolicyHolder().getMarriageStatus().getBytes());
        put.addColumn("policyHolder".getBytes(), "educationalBackground".getBytes(),
                insurancePolicy.getPolicyHolder().getEducationalBackground().getBytes());
        put.addColumn("policyHolder".getBytes(), "phoneNumber".getBytes(),
                insurancePolicy.getPolicyHolder().getPhoneNumber().getBytes());
        put.addColumn("policyHolder".getBytes(), "address".getBytes(),
                insurancePolicy.getPolicyHolder().getAddress().getBytes());
        put.addColumn("policyHolder".getBytes(), "company".getBytes(),
                insurancePolicy.getPolicyHolder().getCompany().getBytes());
        put.addColumn("policyHolder".getBytes(), "position".getBytes(),
                insurancePolicy.getPolicyHolder().getPosition().getBytes());
        put.addColumn("policyHolder".getBytes(), "career".getBytes(),
                insurancePolicy.getPolicyHolder().getCareer().getBytes());
        put.addColumn("policyHolder".getBytes(), "bankCardNumber".getBytes(),
                insurancePolicy.getPolicyHolder().getBankCardNumber().getBytes());
        put.addColumn("policyHolder".getBytes(), "bankName".getBytes(),
                insurancePolicy.getPolicyHolder().getBankName().getBytes());
        put.addColumn("policyHolder".getBytes(), "accountName".getBytes(),
                insurancePolicy.getPolicyHolder().getAccountName().getBytes());

        put.addColumn("insurancePolicyContractStatus".getBytes(), "deliveryDate".getBytes(),
                dateFormat.format(insurancePolicy.getInsurancePolicyContractStatus().getDeliveryDate()).getBytes());
        put.addColumn("insurancePolicyContractStatus".getBytes(), "insureDate".getBytes(),
                dateFormat.format(insurancePolicy.getInsurancePolicyContractStatus().getInsureDate()).getBytes());
        put.addColumn("insurancePolicyContractStatus".getBytes(), "firstExaminationDate".getBytes(),
                dateFormat.format(insurancePolicy.getInsurancePolicyContractStatus().getFirstExaminationDate()).getBytes());
        put.addColumn("insurancePolicyContractStatus".getBytes(), "hesitatePeriodExpiryDate".getBytes(),
                dateFormat.format(insurancePolicy.getInsurancePolicyContractStatus().getHesitatePeriodExpiryDate()).getBytes());
        put.addColumn("insurancePolicyContractStatus".getBytes(), "secondExaminationDate".getBytes(),
                dateFormat.format(insurancePolicy.getInsurancePolicyContractStatus().getSecondExaminationDate()).getBytes());
        put.addColumn("insurancePolicyContractStatus".getBytes(), "expiryDate".getBytes(),
                dateFormat.format(insurancePolicy.getInsurancePolicyContractStatus().getExpiryDate()).getBytes());
        put.addColumn("insurancePolicyContractStatus".getBytes(), "noticeSignDate".getBytes(),
                dateFormat.format(insurancePolicy.getInsurancePolicyContractStatus().getNoticeSignDate()).getBytes());
        put.addColumn("insurancePolicyContractStatus".getBytes(), "receiptSignDate".getBytes(),
                dateFormat.format(insurancePolicy.getInsurancePolicyContractStatus().getReceiptSignDate()).getBytes());
        put.addColumn("insurancePolicyContractStatus".getBytes(), "receiptInputDate".getBytes(),
                dateFormat.format(insurancePolicy.getInsurancePolicyContractStatus().getReceiptInputDate()).getBytes());
        put.addColumn("insurancePolicyContractStatus".getBytes(), "returnVisitDate".getBytes(),
                dateFormat.format(insurancePolicy.getInsurancePolicyContractStatus().getReturnVisitDate()).getBytes());
        put.addColumn("insurancePolicyContractStatus".getBytes(), "postponeDate".getBytes(),
                dateFormat.format(insurancePolicy.getInsurancePolicyContractStatus().getPostponeDate()).getBytes());
        put.addColumn("insurancePolicyContractStatus".getBytes(), "rejectDate".getBytes(),
                dateFormat.format(insurancePolicy.getInsurancePolicyContractStatus().getRejectDate()).getBytes());
        put.addColumn("insurancePolicyContractStatus".getBytes(), "refundDate".getBytes(),
                dateFormat.format(insurancePolicy.getInsurancePolicyContractStatus().getRefundDate()).getBytes());
        put.addColumn("insurancePolicyContractStatus".getBytes(), "cancelDate".getBytes(),
                dateFormat.format(insurancePolicy.getInsurancePolicyContractStatus().getCancelDate()).getBytes());
        put.addColumn("insurancePolicyContractStatus".getBytes(), "expiredDate".getBytes(),
                dateFormat.format(insurancePolicy.getInsurancePolicyContractStatus().getExpiredDate()).getBytes());
        put.addColumn("insurancePolicyContractStatus".getBytes(), "resumeDate".getBytes(),
                dateFormat.format(insurancePolicy.getInsurancePolicyContractStatus().getResumeDate()).getBytes());

        put.addColumn("insurancePlan".getBytes(), "insuranceTypeCode".getBytes(),
                insurancePolicy.getInsurancePlan().getInsuranceTypeCode().getBytes());
        put.addColumn("insurancePlan".getBytes(), "productSpecification".getBytes(),
                insurancePolicy.getInsurancePlan().getProductSpecification().getBytes());
        put.addColumn("insurancePlan".getBytes(), "underlyingAssets".getBytes(),
                insurancePolicy.getInsurancePlan().getUnderlyingAssets().getBytes());
        put.addColumn("insurancePlan".getBytes(), "insurancePremium".getBytes(),
                String.valueOf(insurancePolicy.getInsurancePlan().getInsurancePremium()).getBytes());
        put.addColumn("insurancePlan".getBytes(), "insuranceAmount".getBytes(),
                String.valueOf(insurancePolicy.getInsurancePlan().getInsuranceAmount()).getBytes());
        put.addColumn("insurancePlan".getBytes(), "commissionIncome".getBytes(),
                String.valueOf(insurancePolicy.getInsurancePlan().getCommissionIncome()).getBytes());
        put.addColumn("insurancePlan".getBytes(), "brokeragePay".getBytes(),
                String.valueOf(insurancePolicy.getInsurancePlan().getBrokeragePay()).getBytes());
        put.addColumn("insurancePlan".getBytes(), "acceptInsuranceDate".getBytes(),
                dateFormat.format(insurancePolicy.getInsurancePlan().getAcceptInsuranceDate()).getBytes());
        put.addColumn("insurancePlan".getBytes(), "beginValidDate".getBytes(),
                dateFormat.format(insurancePolicy.getInsurancePlan().getBeginValidDate()).getBytes());
        put.addColumn("insurancePlan".getBytes(), "payDate".getBytes(),
                dateFormat.format(insurancePolicy.getInsurancePlan().getPayDate()).getBytes());
        put.addColumn("insurancePlan".getBytes(), "verifyAccountDate".getBytes(),
                dateFormat.format(insurancePolicy.getInsurancePlan().getVerifyAccountDate()).getBytes());
        put.addColumn("insurancePlan".getBytes(), "giveOutBrokerageDate".getBytes(),
                dateFormat.format(insurancePolicy.getInsurancePlan().getGiveOutBrokerageDate()).getBytes());
        put.addColumn("insurancePlan".getBytes(), "settleBrokerageDate".getBytes(),
                dateFormat.format(insurancePolicy.getInsurancePlan().getSettleBrokerageDate()).getBytes());
        put.addColumn("insurancePlan".getBytes(), "reportDate".getBytes(),
                dateFormat.format(insurancePolicy.getInsurancePlan().getReportDate()).getBytes());

        put.addColumn("videoMaterials".getBytes(), "insureMaterials".getBytes(),
                insurancePolicy.getVideoMaterials().getInsureMaterials().getBytes());
        put.addColumn("videoMaterials".getBytes(), "doubleRecordMaterials".getBytes(),
                insurancePolicy.getVideoMaterials().getDoubleRecordMaterials().getBytes());
        put.addColumn("videoMaterials".getBytes(), "insuranceContract".getBytes(),
                insurancePolicy.getVideoMaterials().getInsuranceContract().getBytes());
        put.addColumn("videoMaterials".getBytes(), "receipt".getBytes(),
                insurancePolicy.getVideoMaterials().getReceipt().getBytes());
        put.addColumn("videoMaterials".getBytes(), "cancelApplicant".getBytes(),
                insurancePolicy.getVideoMaterials().getCancelApplicant().getBytes());
        put.addColumn("videoMaterials".getBytes(), "refundApplicant".getBytes(),
                insurancePolicy.getVideoMaterials().getRefundApplicant().getBytes());
        put.addColumn("videoMaterials".getBytes(), "assureApplicant".getBytes(),
                insurancePolicy.getVideoMaterials().getAssureApplicant().getBytes());
        put.addColumn("videoMaterials".getBytes(), "customerNotice".getBytes(),
                insurancePolicy.getVideoMaterials().getCustomerNotice().getBytes());
        put.addColumn("videoMaterials".getBytes(), "other".getBytes(),
                insurancePolicy.getVideoMaterials().getOther().getBytes());

        put.addColumn("changeHistories".getBytes(), "changeHistories".getBytes(),
                JSONObject.toJSONString(insurancePolicy.getChangeHistories()).getBytes());

        // 向hbase保单表里写入一条保单put请求
        System.out.println("向hbase保单表里写入一条保单put请求......");
        table.put(put);

        // 从hbase里查询刚刚写入的保单验证一下
        Get get = new Get(insurancePolicy.getInsurancePolicyId().getBytes());
        Result result = table.get(get);
        byte[] row = result.getRow();
        System.out.println("向hbase保单表中写入保单成功，rowkey为：" + new String(row));
    }

    /**
     * 从hbase中查询保单
     * @param insurancePolicyId
     */
    public void queryInsurancePolicyFromHBase(String insurancePolicyId) throws Exception {
        HBaseClient hBaseClient = HBaseClient.getInstance();
        Connection connection = hBaseClient.getConnection();
        Table table = connection.getTable(TableName.valueOf("InsurancePolicy"));

        Get get = new Get(insurancePolicyId.getBytes());
        Result result = table.get(get);

        List<String> families = new ArrayList<String>();
        families.add("typeMode");
        families.add("insuranceContract");
        families.add("policyHolder");
        families.add("insurancePolicyContractStatus");
        families.add("insurancePlan");
        families.add("videoMaterials");
        families.add("changeHistories");

        for(String family : families) {
            System.out.println("列族：" + family);
            Map<byte[], byte[]> familyMap = result.getFamilyMap(family.getBytes());

            for(Map.Entry<byte[], byte[]> columnEntry : familyMap.entrySet()) {
                System.out.println(new String(columnEntry.getKey()) + ": " + new String((columnEntry.getValue())));
            }
            System.out.println("============================");
        }
    }

}
